package com.atguigu.day07;

import com.atguigu.day02.Example3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class Example04 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new Example3.MySource())
                .addSink(new RichSinkFunction<Example3.Event>() {
                    private Connection conn ;
                    private PreparedStatement insertStat;
                    private PreparedStatement updateStat;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/sensor", "root", "123456");
                        insertStat = conn.prepareStatement("insert into clicks (name, url) values (?,?)");
                        updateStat = conn.prepareStatement("update clicks set url = ?  where name =?");
                    }

                    @Override
                    public void invoke(Example3.Event value, Context context) throws Exception {
                        super.invoke(value, context);
                        updateStat.setString(1,value.url);
                        updateStat.setString(2,value.name);
                        updateStat.execute();
                        if (updateStat.getUpdateCount()==0){
                            insertStat.setString(1,value.name);
                            insertStat.setString(2,value.url);
                            insertStat.execute();
                        }
                    }

                    @Override
                    public void close() throws Exception {
                        insertStat.close();
                        updateStat.close();
                        conn.close();
                    }
                });


        env.execute();
    }
}
